/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月24日 上午9:46:22
*/
package cn.com.guage.flink.window;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/** 
* @version:1.0.1
* @Description: countWindow用法
* @author: yangdechao
* @date: datedate 2021年11月24日 上午9:46:22
*/
public class CountWindowsTest2 {
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		// 如果不指定时间的话，默认是 ProcessingTime，但是如果指定为事件事件的话，需要事件中带有时间或者添加时间水印
//	    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		/**
		 * 输入数据格式如:
		 * a 10
		 * b 20
		 * c 30
		 * a 20
		 * 
		 */
		DataStreamSource<String> data = env.socketTextStream("192.168.0.128", 9528);
		data.map(new MapFunction<String, Tuple2<String,Integer>>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 5635590728564442978L;

			@Override
			public Tuple2<String, Integer> map(String value) throws Exception {
				String[] splits = value.split(" ");
				return Tuple2.of(splits[0],Integer.parseInt(splits[1]));
			}
		}).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = -5236084965539684057L;

			@Override
			public String getKey(Tuple2<String, Integer> value) throws Exception {
				return value.f0;
			}

		}).countWindow(3,2).sum(1).print();
		env.execute("CountWindowsTest");
	}
}
